/*
 * @Author: wuqingchun
 * @Date: 2024-07-01 14:32:34
 * @Description: 
 * @LastEditors: wuqingchun
 * @LastEditTime: 2024-07-08 20:33:24
 */

#include <sys/prctl.h>

#include "arq_sender.h"

CARQSender::CARQSender():
    CARQSocket(0)
{
    
}

CARQSender::~CARQSender()
{
    
}

int32_t CARQSender::Init(SOCKADDR_UN& refPeerAddr, uint32_t uiWndSize)
{
    int32_t iErr = m_objSndWnd.Init(uiWndSize);
    if (iErr == 0)
    {
        iErr = CARQSocket::Init(refPeerAddr);
    }

    return iErr;
}

int32_t CARQSender::Start()
{
    m_bRecvRuning = true;
    m_objRecvThread = std::thread(&CARQSender::__ThreadFuncRecv, this);

    m_bTimerRuning = true;
    m_objTimerThread = std::thread(&CARQSender::__ThreadFuncTimer, this);

    return 0;
}

int32_t CARQSender::Send(const void* pData, size_t ulLen)
{
    if (m_objSndWnd.IsWndFull())
    {
        // 发送端需要停止发送，等待接收端确认
        std::unique_lock<std::mutex> objUL(m_objMtxSnd);
        m_objCVSnd.wait(objUL);
    }

    // 填充ARQ协议头
    ARQ_HEAD_S* pstHdr = (ARQ_HEAD_S*)pData;
    m_objSndWnd.AddMsg(pstHdr);
    pstHdr->uiAckNo = 0;
    pstHdr->usChksum = 0; //TODO
    pstHdr->usLength = (uint16_t)ulLen;

    int32_t iErr = CARQSocket::Send(pData, ulLen);
    if (iErr == 0)
    {
        // 启动定时器
        m_objTimer.AddTask(10000000, 0, [&](const void* pPrivate){
            uint32_t uiSeqNo = (uint32_t)(u_long)pPrivate;
            this->__RetransmitMsg(uiSeqNo);
        }, (const void*)(u_long)pstHdr->uiSeqNo);

        //printf("Send request successfully, seqno=%u\n", pstHdr->uiSeqNo);
    }

    return iErr;
}
void* CARQSender::Alloc(size_t ulSize)
{
    // 预留消息头
    void* pMem = m_objAllocator.Alloc(sizeof(ARQ_HEAD_S) + ulSize);
    if (pMem != nullptr)
    {
        return (char*)pMem + sizeof(ARQ_HEAD_S);
    }

    return nullptr;
}

void CARQSender::GetStatistics()
{
    m_objSndWnd.GetStatistics();
}

void CARQSender::__ThreadFuncRecv()
{
    char acBuf[1024];
    ssize_t lRes;


    (void)prctl(PR_SET_NAME, "tx_recv");

    while (m_bRecvRuning)
    {
        lRes = CARQSocket::Recv(acBuf, 1024);
        if (lRes >= sizeof(ARQ_HEAD_S))
        {
            ARQ_HEAD_S* pstHdr = (ARQ_HEAD_S*)acBuf;
            __HandleAck(*pstHdr);
        }
    }
}

int32_t CARQSender::__HandleAck(ARQ_HEAD_S& refHdr)
{
    m_objSndWnd.OnAck(refHdr.uiAckNo);
    if (m_objSndWnd.OnAck(refHdr.uiAckNo))
    {
        // 删除重传定时任务
        m_objTimer.RemoveTask([](const void* pData, const void* pKey)->bool{
            uint32_t uiSeqNo = (uint32_t)(long)pData;
            ARQ_HEAD_S* pstKey = (ARQ_HEAD_S*)pKey;

            if (uiSeqNo <= pstKey->uiSeqNo)
            {
                return true;
            }

            return false;
        }, (const void*)&refHdr);

        // 通知发送窗口释放出一个位置
        std::unique_lock<std::mutex> objUL(m_objMtxSnd);
        m_objCVSnd.notify_one();
    }

    return 0;
}

int32_t CARQSender::__SendHeartBeat()
{

}

void CARQSender::__ThreadFuncTimer()
{
    (void)prctl(PR_SET_NAME, "tx_timer");

    while (m_bTimerRuning)
    {
        m_objTimer.Run();
        std::this_thread::sleep_for(std::chrono::microseconds(1));
    }
}

int32_t CARQSender::__RetransmitMsg(uint32_t uiSeqNo)
{
    int32_t iErr = -1;
    ARQ_HEAD_S* pstHdr = m_objSndWnd.GetMsg(uiSeqNo);
    if (pstHdr != nullptr)
    {
        iErr = CARQSocket::Send(pstHdr, pstHdr->usLength);
        if (iErr == 0)
        {
            // 启动重传定时任务
            m_objTimer.AddTask(10000000, 0, [&](const void* pPrivate){
                uint32_t uiSeqNo = (uint32_t)(u_long)pPrivate;
                this->__RetransmitMsg(uiSeqNo);
            }, (const void*)(u_long)pstHdr->uiSeqNo);

            //printf("Retransmitted request successfully, seqno=%u\n", pstHdr->uiSeqNo);
        }
    }

    return iErr;
}